package com.xlg.file.center.utils;

import lombok.extern.slf4j.Slf4j;

import java.util.Map;
import java.util.concurrent.*;

/**
 * @author: 18310740596@163.com
 * @date: 2023-07-14 22:43:57
 * @version: v1.0
 * @describe: class responsibility
 */
@Slf4j
public class TaskUtil {
    /**
     *  任务处理器缓存
     */
    private static final Map<String, TaskHandler> TASK_CACHE = new ConcurrentHashMap<>(128);

    /**
     *  结果集缓存
     */
    private static final Map<String, CompletableFuture<TaskResult>> RESULT_CACHE = new ConcurrentHashMap<>(128);

    /**
     *  任务等待队列
     */
    private static final BlockingQueue<Task> TASK_WAITING_QUEUE = new ArrayBlockingQueue<>(1024);

    /**
     *  同时只能有10文件创建动作， 其余的创建动作在阻塞队列等待令牌。
     */
    private static final Semaphore BUCKET = new Semaphore(10);

    /**
     *  是否启动等待任务处理器
     */
    public static volatile boolean isStartWaitingTaskHandler = false;


    /**
     *  文件创建
     *
     * @param task 创建任务信息
     * @throws Exception 异常处理
     */
    public static void createFile(Task task) throws Exception {
        log.info("{}.createFile() 创建文件请求参数:{}", TaskUtil.class.getSimpleName(), task.getUniqNo());

        if (!RESULT_CACHE.containsKey(task.getUniqNo())) {
            RESULT_CACHE.put(task.getUniqNo(), new CompletableFuture<>());
        }
        // 如果已经获得了令牌,则直接加入任务进行处理
        if (TASK_CACHE.containsKey(task.getUniqNo())) {
            TASK_CACHE.get(task.getUniqNo()).registerTask(task);
        } else {
            boolean bucket = BUCKET.tryAcquire(1, 3, TimeUnit.SECONDS);

            if (bucket) {
                TASK_CACHE.computeIfAbsent(task.getUniqNo(), k -> new TaskHandler(BUCKET)).registerTask(task);
            } else {
                addWaitingQueue(task);
            }
        }
    }

    /**
     *  结果集填充
     *
     * @param uniqNo 唯一批次号
     * @return 响应结果
     */
    public static CompletableFuture<TaskResult> resultFill(String uniqNo) {
        return RESULT_CACHE.get(uniqNo);
    }

    /**
     *  等待任务处理结果
     *
     * @param uniqNo 任务唯一编号
     * @return 文件处理结果
     */
    public static TaskResult waitingHandlerTaskResult(String uniqNo) {
        return RESULT_CACHE.remove(uniqNo).join();
    }

    /**
     *  任务处理器处理完成时，移除处理器
     *
     * @param uniqNo 任务唯一编号
     */
    public static void handlerRemove(String uniqNo) {
        TASK_CACHE.remove(uniqNo);
    }

    private static void addWaitingQueue(Task task) throws Exception {
        if (!isStartWaitingTaskHandler) {
            doStart();
        }
        boolean offer = TASK_WAITING_QUEUE.offer(task);

        if (!offer) {
            throw new Exception("目前任务数已满,请稍后再试!");
        }
    }

    private static void doStart() {
        if (!isStartWaitingTaskHandler) {
            synchronized (TaskUtil.class) {
                if (!isStartWaitingTaskHandler) {
                    CompletableFuture.runAsync(TaskUtil::taskListener);
                    isStartWaitingTaskHandler = true;
                }
            }
        }
    }

    private static void taskListener() {
        while (true) {
            try {
                boolean bucket = BUCKET.tryAcquire(1, 3, TimeUnit.SECONDS);
                if (bucket) {
                    Task task = TASK_WAITING_QUEUE.take();

                    if (TASK_CACHE.containsKey(task.getUniqNo())) {
                        TASK_CACHE.get(task.getUniqNo()).registerTask(task);
                    } else {
                        TASK_CACHE.computeIfAbsent(task.getUniqNo(), k -> new TaskHandler(BUCKET)).registerTask(task);
                    }
                } else {
                    log.info("当前生成文件请求过多,正在排队中...");
                }
            } catch (InterruptedException ignore) {
            }
        }
    }
}
